package org.lxy.config;

import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.lxy.core.mq.AbstractRocketConsumer;
import org.lxy.domain.config.RocketMqProperties;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Map;

/**
 * @author 卢心怡
 */
@Configuration
@EnableConfigurationProperties({RocketMqProperties.class})
@ConditionalOnProperty(prefix = "super.rocketmq", value = "isEnable", havingValue = "true",matchIfMissing = true)
@Slf4j
@Setter
public class RocketMQConfig implements ApplicationRunner {

    RocketMqProperties rocketMqProperties;

    ApplicationContext applicationContext;

    public RocketMQConfig(RocketMqProperties rocketMqProperties, ApplicationContext applicationContext) {
        this.rocketMqProperties = rocketMqProperties;
        this.applicationContext = applicationContext;
    }

    @Bean
    public DefaultMQProducer defaultMQProducer(){
        if (StringUtils.isEmpty(rocketMqProperties.getGroupName()) || StringUtils.isEmpty(rocketMqProperties.getNamesrvAddr())) {
            throw new RuntimeException("rocket 一些重要属性为空!");
        }
        DefaultMQProducer producer;
        producer = new DefaultMQProducer(rocketMqProperties.getGroupName());

        producer.setNamesrvAddr(rocketMqProperties.getNamesrvAddr());
        // producer.setCreateTopicKey("AUTO_CREATE_TOPIC_KEY");

        // 如果需要同一个jvm中不同的producer往不同的mq集群发送消息，需要设置不同的instanceName
        // producer.setInstanceName(instanceName);
        producer.setMaxMessageSize(rocketMqProperties.getProducerMaxMessageSize());
        producer.setSendMsgTimeout(rocketMqProperties.getProducerSendMsgTimeout());
        // 如果发送消息失败，设置重试次数，默认为2次
        producer.setRetryTimesWhenSendFailed(rocketMqProperties.getProducerRetryTimesWhenSendFailed());

        try {
            producer.start();
            log.info("producer is start ! groupName:{},namesrvAddr:{}", rocketMqProperties.getGroupName(),
                    rocketMqProperties.getNamesrvAddr());
        } catch (MQClientException e) {
            log.error(String.format("producer is error {}", e.getMessage(), e));
            throw new RuntimeException(e);
        }
        return producer;
    }


    @Bean
    public DefaultMQPushConsumer defaultMQPushConsumer(){
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(rocketMqProperties.getGroupName());
        defaultMQPushConsumer.setNamesrvAddr(rocketMqProperties.getNamesrvAddr());
        return defaultMQPushConsumer;
    }

    /**
     * 通过消费者信息创建消费者
     *
     * @param arc
     */
    public void createConsumer(AbstractRocketConsumer arc, DefaultMQPushConsumer consumer) {
        consumer.setNamesrvAddr(this.rocketMqProperties.getNamesrvAddr());
        consumer.setConsumeThreadMin(this.rocketMqProperties.getConsumerConsumeThreadMin());
        consumer.setConsumeThreadMax(this.rocketMqProperties.getConsumerConsumeThreadMax());
        consumer.registerMessageListener(arc.getMessageListener());
        /**
         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费 如果非第一次启动，那么按照上次消费的位置继续消费
         */
        // consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        /**
         * 设置消费模型，集群还是广播，默认为集群
         */
        // consumer.setMessageModel(MessageModel.CLUSTERING);

        /**
         * 设置一次消费消息的条数，默认为1条
         */
        consumer.setConsumeMessageBatchMaxSize(this.rocketMqProperties.getConsumerConsumeMessageBatchMaxSize());
        try {
            consumer.subscribe(arc.getTopics(), arc.getTags());
            consumer.start();
            arc.setMqPushConsumer(consumer);
        } catch (MQClientException e) {
            log.error("info consumer title {}", arc.getConsumerTitel(), e);
        }

    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        Map<String, AbstractRocketConsumer> beansOfType = applicationContext.getBeansOfType(AbstractRocketConsumer.class);
        DefaultMQPushConsumer defaultMQPushConsumer = (DefaultMQPushConsumer) applicationContext.getBean("defaultMQPushConsumer");
        beansOfType.values().stream().forEach(e ->{
            e.init();
            createConsumer(e,defaultMQPushConsumer);
            log.info("init success consumer title {} , toips {} , tags {}", e.getConsumerTitel(),
                    e.getTopics(), e.getTags());
        });
    }
}
